package com.reactim.message.connection;

import com.reactim.message.config.ConnectionConfig;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 连接监控服务
 * 负责监控连接状态、执行心跳检测和生成连接统计报告
 */
@Service
@Slf4j
public class ConnectionMonitoringService {

    private final ConnectionManager connectionManager;
    private final ConnectionConfig connectionConfig;

    public ConnectionMonitoringService(
            @Qualifier("cachedConnectionManager") ConnectionManager connectionManager,
            ConnectionConfig connectionConfig) {
        this.connectionManager = connectionManager;
        this.connectionConfig = connectionConfig;
    }
    
    // 连接统计数据
    private final Map<String, AtomicLong> connectionStats = new ConcurrentHashMap<>();
    
    // 实例连接分布
    private final Map<String, AtomicLong> instanceConnectionCounts = new ConcurrentHashMap<>();
    
    // 连接事件历史（最近100个事件）
    private final List<ConnectionEvent> recentEvents = new ArrayList<>(100);
    
    // 心跳检测结果
    private final Map<Long, HeartbeatStatus> heartbeatStatuses = new ConcurrentHashMap<>();
    
    /**
     * 初始化统计计数器
     */
    public void initializeStats() {
        // 初始化基本统计计数器
        connectionStats.put("totalConnections", new AtomicLong(0));
        connectionStats.put("activeConnections", new AtomicLong(0));
        connectionStats.put("connectionsCreated", new AtomicLong(0));
        connectionStats.put("connectionsRemoved", new AtomicLong(0));
        connectionStats.put("heartbeatsSent", new AtomicLong(0));
        connectionStats.put("heartbeatsReceived", new AtomicLong(0));
        connectionStats.put("heartbeatsFailed", new AtomicLong(0));
        connectionStats.put("inactiveConnectionsRemoved", new AtomicLong(0));
        
        // 更新当前连接总数
        updateTotalConnections();
    }
    
    /**
     * 更新总连接数
     */
    private void updateTotalConnections() {
        connectionManager.getOnlineUserCount()
            .subscribe(count -> {
                connectionStats.get("totalConnections").set(count);
                connectionStats.get("activeConnections").set(count);
                log.debug("更新总连接数: {}", count);
            }, error -> {
                log.error("获取总连接数失败: {}", error.getMessage());
            });
    }
    
    /**
     * 记录连接创建事件
     * 
     * @param userId 用户ID
     * @param instanceId 实例ID
     */
    public void recordConnectionCreated(Long userId, String instanceId) {
        connectionStats.get("connectionsCreated").incrementAndGet();
        connectionStats.get("totalConnections").incrementAndGet();
        connectionStats.get("activeConnections").incrementAndGet();
        
        // 更新实例连接计数
        instanceConnectionCounts.computeIfAbsent(instanceId, k -> new AtomicLong(0)).incrementAndGet();
        
        // 记录事件
        addConnectionEvent(new ConnectionEvent(
            ConnectionEventType.CONNECTED,
            userId,
            instanceId,
            System.currentTimeMillis(),
            null
        ));
        
        log.debug("记录连接创建: userId={}, instanceId={}", userId, instanceId);
    }
    
    /**
     * 记录连接移除事件
     * 
     * @param userId 用户ID
     * @param instanceId 实例ID
     * @param reason 移除原因
     */
    public void recordConnectionRemoved(Long userId, String instanceId, String reason) {
        connectionStats.get("connectionsRemoved").incrementAndGet();
        connectionStats.get("totalConnections").decrementAndGet();
        connectionStats.get("activeConnections").decrementAndGet();
        
        // 更新实例连接计数
        AtomicLong instanceCount = instanceConnectionCounts.get(instanceId);
        if (instanceCount != null) {
            instanceCount.decrementAndGet();
        }
        
        // 记录事件
        addConnectionEvent(new ConnectionEvent(
            ConnectionEventType.DISCONNECTED,
            userId,
            instanceId,
            System.currentTimeMillis(),
            reason
        ));
        
        // 移除心跳状态
        heartbeatStatuses.remove(userId);
        
        log.debug("记录连接移除: userId={}, instanceId={}, reason={}", userId, instanceId, reason);
    }
    
    /**
     * 添加连接事件到历史记录
     * 
     * @param event 连接事件
     */
    private synchronized void addConnectionEvent(ConnectionEvent event) {
        // 如果历史记录已满，移除最旧的事件
        if (recentEvents.size() >= 100) {
            recentEvents.remove(0);
        }
        recentEvents.add(event);
    }
    
    /**
     * 执行心跳检测
     * 定期检查所有连接的活跃状态
     */
    @Scheduled(fixedDelayString = "${connection.monitoring.heartbeat-interval-ms:30000}")
    public void performHeartbeatCheck() {
        log.debug("开始执行心跳检测");
        
        long inactivityThreshold = System.currentTimeMillis() - 
            connectionConfig.getMonitoring().getInactivityThresholdMs();
        
        // 获取所有连接
        connectionManager.getConnectionsByInstance("*")
            .flatMapMany(Flux::fromIterable)
            .flatMap(connection -> {
                Long userId = connection.getInfo().getUserId();
                connectionStats.get("heartbeatsSent").incrementAndGet();
                
                // 获取连接详情
                return connectionManager.getConnection(userId)
                    .flatMap(details -> {
                        // 检查最后活跃时间
                        if (details.getLastActiveTime() < inactivityThreshold) {
                            // 连接不活跃，记录状态
                            heartbeatStatuses.put(userId, new HeartbeatStatus(
                                userId,
                                HeartbeatResult.INACTIVE,
                                System.currentTimeMillis(),
                                "连接不活跃超过阈值"
                            ));
                            
                            // 如果启用了自动清理，则移除不活跃连接
                            if (connectionConfig.getMonitoring().isAutoRemoveInactive()) {
                                log.info("移除不活跃连接: userId={}, lastActiveTime={}", 
                                    userId, Instant.ofEpochMilli(details.getLastActiveTime()));
                                
                                connectionStats.get("inactiveConnectionsRemoved").incrementAndGet();
                                
                                return connectionManager.removeConnection(userId)
                                    .map(removed -> {
                                        if (removed) {
                                            recordConnectionRemoved(
                                                userId, 
                                                details.getInfo().getInstanceId(), 
                                                "心跳检测：连接不活跃"
                                            );
                                        }
                                        return details;
                                    });
                            }
                        } else {
                            // 连接活跃，发送心跳
                            return sendHeartbeat(userId, details)
                                .map(result -> {
                                    // 记录心跳结果
                                    heartbeatStatuses.put(userId, new HeartbeatStatus(
                                        userId,
                                        result ? HeartbeatResult.SUCCESS : HeartbeatResult.FAILED,
                                        System.currentTimeMillis(),
                                        result ? "心跳成功" : "心跳失败"
                                    ));
                                    
                                    if (result) {
                                        connectionStats.get("heartbeatsReceived").incrementAndGet();
                                    } else {
                                        connectionStats.get("heartbeatsFailed").incrementAndGet();
                                        
                                        // 如果启用了自动清理，则移除心跳失败的连接
                                        if (connectionConfig.getMonitoring().isAutoRemoveFailedHeartbeat()) {
                                            log.info("移除心跳失败的连接: userId={}", userId);
                                            
                                            connectionManager.removeConnection(userId)
                                                .subscribe(removed -> {
                                                    if (removed) {
                                                        recordConnectionRemoved(
                                                            userId, 
                                                            details.getInfo().getInstanceId(), 
                                                            "心跳检测：心跳失败"
                                                        );
                                                    }
                                                });
                                        }
                                    }
                                    
                                    return details;
                                });
                        }
                        
                        return Mono.just(details);
                    })
                    .onErrorResume(error -> {
                        log.error("心跳检测出错: userId={}, error={}", userId, error.getMessage());
                        return Mono.empty();
                    });
            })
            .collectList()
            .subscribe(
                result -> log.debug("心跳检测完成: 检查了{}个连接", result.size()),
                error -> log.error("心跳检测过程出错: {}", error.getMessage())
            );
    }
    
    /**
     * 发送心跳消息
     * 
     * @param userId 用户ID
     * @param details 连接详情
     * @return 心跳结果
     */
    private Mono<Boolean> sendHeartbeat(Long userId, ConnectionDetails details) {
        // 如果配置为不发送实际心跳，则只更新活跃时间
        if (!connectionConfig.getMonitoring().isSendActualHeartbeat()) {
            return connectionManager.updateLastActiveTime(userId);
        }
        
        // 发送实际心跳消息
        return connectionManager.sendMessageToUser(userId, "heartbeat", Map.of("timestamp", System.currentTimeMillis()))
            .then(Mono.just(true))
            .onErrorReturn(false);
    }
    
    /**
     * 生成连接统计报告
     * 
     * @return 统计报告
     */
    public ConnectionStatisticsReport generateStatisticsReport() {
        Map<String, Long> stats = new HashMap<>();
        connectionStats.forEach((key, value) -> stats.put(key, value.get()));
        
        Map<String, Long> instanceStats = new HashMap<>();
        instanceConnectionCounts.forEach((key, value) -> instanceStats.put(key, value.get()));
        
        // 计算心跳统计
        long totalHeartbeats = stats.getOrDefault("heartbeatsSent", 0L);
        long successfulHeartbeats = stats.getOrDefault("heartbeatsReceived", 0L);
        long failedHeartbeats = stats.getOrDefault("heartbeatsFailed", 0L);
        
        double heartbeatSuccessRate = totalHeartbeats > 0 
            ? (double) successfulHeartbeats / totalHeartbeats 
            : 0.0;
        
        return ConnectionStatisticsReport.builder()
            .timestamp(System.currentTimeMillis())
            .connectionStats(stats)
            .instanceConnectionCounts(instanceStats)
            .recentEvents(new ArrayList<>(recentEvents))
            .heartbeatSuccessRate(heartbeatSuccessRate)
            .build();
    }
    
    /**
     * 获取所有心跳状态
     * 
     * @return 心跳状态映射
     */
    public Map<Long, HeartbeatStatus> getHeartbeatStatuses() {
        return new HashMap<>(heartbeatStatuses);
    }
    
    /**
     * 获取指定用户的心跳状态
     * 
     * @param userId 用户ID
     * @return 心跳状态
     */
    public HeartbeatStatus getUserHeartbeatStatus(Long userId) {
        return heartbeatStatuses.get(userId);
    }
    
    /**
     * 重置统计数据
     */
    public void resetStatistics() {
        connectionStats.forEach((key, value) -> {
            // 保留当前连接总数和活跃连接数
            if (!key.equals("totalConnections") && !key.equals("activeConnections")) {
                value.set(0);
            }
        });
        
        log.info("连接统计数据已重置");
    }
}

/**
 * 连接事件类型
 */
enum ConnectionEventType {
    CONNECTED,
    DISCONNECTED,
    HEARTBEAT_SUCCESS,
    HEARTBEAT_FAILED
}

/**
 * 连接事件
 */
class ConnectionEvent {
    private final ConnectionEventType type;
    private final Long userId;
    private final String instanceId;
    private final long timestamp;
    private final String details;
    
    public ConnectionEvent(ConnectionEventType type, Long userId, String instanceId, long timestamp, String details) {
        this.type = type;
        this.userId = userId;
        this.instanceId = instanceId;
        this.timestamp = timestamp;
        this.details = details;
    }
    
    public ConnectionEventType getType() {
        return type;
    }
    
    public Long getUserId() {
        return userId;
    }
    
    public String getInstanceId() {
        return instanceId;
    }
    
    public long getTimestamp() {
        return timestamp;
    }
    
    public String getDetails() {
        return details;
    }
}

/**
 * 心跳结果
 */
enum HeartbeatResult {
    SUCCESS,
    FAILED,
    INACTIVE
}

/**
 * 心跳状态
 */
class HeartbeatStatus {
    private final Long userId;
    private final HeartbeatResult result;
    private final long timestamp;
    private final String message;
    
    public HeartbeatStatus(Long userId, HeartbeatResult result, long timestamp, String message) {
        this.userId = userId;
        this.result = result;
        this.timestamp = timestamp;
        this.message = message;
    }
    
    public Long getUserId() {
        return userId;
    }
    
    public HeartbeatResult getResult() {
        return result;
    }
    
    public long getTimestamp() {
        return timestamp;
    }
    
    public String getMessage() {
        return message;
    }
}

/**
 * 连接统计报告
 */
@lombok.Builder
class ConnectionStatisticsReport {
    private final long timestamp;
    private final Map<String, Long> connectionStats;
    private final Map<String, Long> instanceConnectionCounts;
    private final List<ConnectionEvent> recentEvents;
    private final double heartbeatSuccessRate;
    
    public long getTimestamp() {
        return timestamp;
    }
    
    public Map<String, Long> getConnectionStats() {
        return connectionStats;
    }
    
    public Map<String, Long> getInstanceConnectionCounts() {
        return instanceConnectionCounts;
    }
    
    public List<ConnectionEvent> getRecentEvents() {
        return recentEvents;
    }
    
    public double getHeartbeatSuccessRate() {
        return heartbeatSuccessRate;
    }
}